package io.kiki.sba.registry.server.metadata.multi.cluster;

import com.google.common.collect.Sets;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import lombok.extern.slf4j.Slf4j;

import java.util.Set;

@Slf4j
final class SlotMapWatcher extends WakeUpLoopExecuteTask {

    private final MultiClusterSlotsSyncerImpl multiClusterSlotMapSyncer;

    public SlotMapWatcher(MultiClusterSlotsSyncerImpl multiClusterSlotMapSyncer) {
        this.multiClusterSlotMapSyncer = multiClusterSlotMapSyncer;
    }

    @Override
    public void _execute_() {
        try {
            // if need reload sync info from db
            if (multiClusterSlotMapSyncer.needReloadConfig()) {
                multiClusterSlotMapSyncer.getExecutorManager().getMultiClusterConfigReloadExecutor().execute(() -> multiClusterSlotMapSyncer.remoteClusterMetaExchanger.refreshDataCenterToMultiClusterSyncInfoMap());
                MultiClusterSlotsSyncerImpl.LAST_REFRESH_CONFIG_TS = System.currentTimeMillis();
            }
        } catch (Throwable t) {
            log.error("refresh multi cluster config error.", t);
        }

        if (!multiClusterSlotMapSyncer.metadataLeaderService.amILeader()) {
            return;
        }

        // remoteDataCenters is save in MultiClusterSyncInfo.remoteDataCenter, just use to print log,
        // not clusterId
        Set<String> remoteDataCenters = multiClusterSlotMapSyncer.remoteClusterMetaExchanger.getAllRemoteClusters();
        Sets.SetView<String> removes = Sets.difference(multiClusterSlotMapSyncer.dataCenterToRemoteClusterSlotStateMap.keySet(), remoteDataCenters);
        for (String remove : removes) {
            MultiClusterSlotsSyncerImpl.RemoteClusterSlotState remoteClusterSlotState = multiClusterSlotMapSyncer.dataCenterToRemoteClusterSlotStateMap.remove(remove);
            log.info("remove dataCenter: {} sync info: {}.", remove, remoteClusterSlotState);
        }

        for (String dataCenter : remoteDataCenters) {
            MultiClusterSlotsSyncerImpl.RemoteClusterSlotState remoteClusterSlotState = multiClusterSlotMapSyncer.dataCenterToRemoteClusterSlotStateMap.computeIfAbsent(dataCenter, k -> new MultiClusterSlotsSyncerImpl.RemoteClusterSlotState());

            // check if exceed max fail count
            if (checkSyncFailCount(remoteClusterSlotState.getFailCount())) {
                multiClusterSlotMapSyncer.executorManager.getRemoteSlotSyncerExecutor().execute(dataCenter, () -> {
                    remoteClusterSlotState.initFailCount();
                    multiClusterSlotMapSyncer.remoteClusterMetaExchanger.resetLeader(dataCenter);
                });
                continue;
            }

            // check if need to do sync
            if (multiClusterSlotMapSyncer.needSync(remoteClusterSlotState.task)) {
                RemoteClusterSlotSyncTask remoteClusterSlotSyncTask = new RemoteClusterSlotSyncTask(multiClusterSlotMapSyncer, dataCenter, remoteClusterSlotState.slots.getEpoch());
                remoteClusterSlotState.task = multiClusterSlotMapSyncer.executorManager.getRemoteSlotSyncerExecutor().execute(dataCenter, remoteClusterSlotSyncTask);
                log.info("commit sync task:{}", remoteClusterSlotSyncTask);
            }
        }
    }

    private boolean checkSyncFailCount(long failCount) {
        if (failCount >= MultiClusterSlotsSyncerImpl.MAX_SYNC_FAIL_COUNT) {
            log.error("sync failed [{}] times, prepare to reset leader from rest api.", failCount);
            return true;
        }
        return false;
    }

    @Override
    public int getWaitingMillis() {
        return 200;
    }
}
